package com.mini.grpc.transport.netty;

import com.mini.grpc.common.Status;
import com.mini.grpc.common.Metadata;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Netty客户端处理器，处理HTTP/2帧
 */
public class NettyClientHandler extends Http2ConnectionHandler implements Http2FrameListener {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);
    
    private final ConcurrentMap<Integer, NettyClientStream<?, ?>> streams = new ConcurrentHashMap<>();
    private ChannelHandlerContext ctx;
    
    public NettyClientHandler(Http2ConnectionDecoder decoder, Http2ConnectionEncoder encoder, Http2Settings initialSettings) {
        super(decoder, encoder, initialSettings);
    }
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        super.handlerAdded(ctx);
        this.ctx = ctx;
    }
    
    public void addStream(int streamId, NettyClientStream<?, ?> stream) {
        streams.put(streamId, stream);
    }
    
    public void removeStream(int streamId) {
        streams.remove(streamId);
    }
    
    public void sendHeaders(int streamId, Http2Headers headers, boolean endStream) {
        // 实现发送头部的逻辑
        logger.debug("Sending headers for stream {}: {}", streamId, headers);
        encoder().writeHeaders(ctx, streamId, headers, 0, endStream, ctx.newPromise());
        ctx.flush();
    }
    
    public void sendData(int streamId, byte[] data, boolean endStream) {
        // 实现发送数据的逻辑
        logger.debug("Sending data for stream {}: {} bytes, endStream={}", streamId, data.length, endStream);
        io.netty.buffer.ByteBuf buffer = ctx.alloc().buffer(data.length);
        buffer.writeBytes(data);
        encoder().writeData(ctx, streamId, buffer, 0, endStream, ctx.newPromise());
        ctx.flush();
    }
    
    public void cancelStream(int streamId) {
        // 实现取消流的逻辑
        logger.debug("Cancelling stream {}", streamId);
        try {
            encoder().writeRstStream(ctx, streamId, 8, ctx.newPromise()); // CANCEL error code
            ctx.flush();
        } catch (Exception e) {
            logger.error("Failed to cancel stream {}", streamId, e);
        } finally {
            removeStream(streamId);
        }
    }
    
    private static int nextStreamIdCounter = 1;
    
    public int nextStreamId() {
        // 生成下一个流ID（客户端使用奇数）
        int streamId = nextStreamIdCounter;
        nextStreamIdCounter += 2; // 客户端流ID必须是奇数
        return streamId;
    }
    
    @Override
    public int onDataRead(ChannelHandlerContext ctx, int streamId, io.netty.buffer.ByteBuf data, int padding, boolean endOfStream) throws Http2Exception {
        logger.debug("=== onDataRead called: streamId={}, dataSize={}, padding={}, endOfStream={} ===", 
                    streamId, data.readableBytes(), padding, endOfStream);
        
        NettyClientStream<?, ?> stream = streams.get(streamId);
        if (stream != null) {
            // 解析gRPC消息格式：压缩标志(1字节) + 消息长度(4字节) + 消息内容
            while (data.isReadable()) {
                // 确保有足够的数据读取头部
                if (data.readableBytes() < 5) {
                    logger.debug("Not enough data for message header, waiting for more data");
                    break;
                }
                
                // 标记读取位置，以便回滚
                int readerIndex = data.readerIndex();
                
                // 读取压缩标志
                boolean compressed = data.readByte() != 0;
                
                // 读取消息长度（大端序）
                int length = data.readInt();
                
                logger.debug("Message header: compressed={}, length={}", compressed, length);
                
                // 确保有足够的数据读取消息体
                if (data.readableBytes() < length) {
                    // 不完整的消息，回滚读取位置并等待更多数据
                    data.readerIndex(readerIndex);
                    logger.debug("Not enough data for message body, waiting for more data");
                    break;
                }
                
                // 读取消息内容
                byte[] messageBytes = new byte[length];
                data.readBytes(messageBytes);
                
                logger.debug("Read message: {} bytes", messageBytes.length);
                
                // 反序列化消息（这里简单地转换为字符串）
                String message = new String(messageBytes, java.nio.charset.StandardCharsets.UTF_8);
                logger.debug("Deserialized message: {}", message);
                
                @SuppressWarnings("unchecked")
                NettyClientStream<Object, Object> rawStream = (NettyClientStream<Object, Object>) stream;
                rawStream.onMessage(message);
            }
            
            if (endOfStream) {
                logger.debug("End of stream for stream {}", streamId);
                stream.onComplete();
                removeStream(streamId);
            }
        } else {
            logger.error("Data received for unknown stream: {}", streamId);
        }
        return data.readableBytes() + padding;
    }
    
    @Override
    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding, boolean endOfStream) throws Http2Exception {
        NettyClientStream<?, ?> stream = streams.get(streamId);
        if (stream != null) {
            Metadata metadata = convertHeaders(headers);
            stream.onHeaders(metadata);
            
            if (endOfStream) {
                stream.onComplete();
                removeStream(streamId);
            }
        }
    }
    
    @Override
    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency, short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
        onHeadersRead(ctx, streamId, headers, padding, endOfStream);
    }
    
    @Override
    public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight, boolean exclusive) throws Http2Exception {
        // 忽略优先级帧
    }
    
    @Override
    public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
        NettyClientStream<?, ?> stream = streams.get(streamId);
        if (stream != null) {
            Status status = Status.CANCELLED.withDescription("Stream reset with error code: " + errorCode);
            stream.onError(status, new Metadata());
            removeStream(streamId);
        }
    }
    
    @Override
    public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
        // 忽略设置确认帧
    }
    
    @Override
    public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
        // 忽略设置帧
    }
    
    @Override
    public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
        // 忽略ping帧
    }
    
    @Override
    public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
        // 忽略ping确认帧
    }
    
    @Override
    public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId, Http2Headers headers, int padding) throws Http2Exception {
        // 忽略推送承诺帧
    }
    
    @Override
    public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, io.netty.buffer.ByteBuf debugData) throws Http2Exception {
        logger.warn("Received GOAWAY frame: lastStreamId={}, errorCode={}", lastStreamId, errorCode);
        // 关闭所有活跃的流
        for (NettyClientStream<?, ?> stream : streams.values()) {
            Status status = Status.UNAVAILABLE.withDescription("Connection going away");
            stream.onError(status, new Metadata());
        }
        streams.clear();
    }
    
    @Override
    public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement) throws Http2Exception {
        // 忽略窗口更新帧
    }
    
    @Override
    public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags, io.netty.buffer.ByteBuf payload) throws Http2Exception {
        // 忽略未知帧
    }
    
    private Metadata convertHeaders(Http2Headers headers) {
        Metadata metadata = new Metadata();
        for (CharSequence name : headers.names()) {
            for (CharSequence value : headers.getAll(name)) {
                if (name.toString().endsWith("-bin")) {
                    // 二进制头
                    metadata.put(Metadata.Key.of(name.toString(), Metadata.BINARY_BYTE_MARSHALLER), 
                               value.toString().getBytes());
                } else {
                    // 文本头
                    metadata.put(Metadata.Key.of(name.toString(), Metadata.ASCII_STRING_MARSHALLER), 
                               value.toString());
                }
            }
        }
        return metadata;
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("Exception in NettyClientHandler", cause);
        // 通知所有活跃的流发生错误
        for (NettyClientStream<?, ?> stream : streams.values()) {
            Status status = Status.fromThrowable(cause);
            stream.onError(status, new Metadata());
        }
        streams.clear();
        super.exceptionCaught(ctx, cause);
    }
} 